package cn.itsource.ymcc.service.impl;

import cn.itsource.ymcc.constants.BaseConstants;
import cn.itsource.ymcc.context.LoginContext;
import cn.itsource.ymcc.domain.KillActivity;
import cn.itsource.ymcc.domain.KillCourse;
import cn.itsource.ymcc.dto.KillParamDto;
import cn.itsource.ymcc.exception.GlobleException;
import cn.itsource.ymcc.mapper.KillActivityMapper;
import cn.itsource.ymcc.mapper.KillCourseMapper;
import cn.itsource.ymcc.service.IKillCourseService;
import cn.itsource.ymcc.to.KilledInfo;
import cn.itsource.ymcc.util.AssertUtil;
import cn.itsource.ymcc.util.StrUtils;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.mapper.EntityWrapper;
import com.baomidou.mybatisplus.mapper.Wrapper;
import com.baomidou.mybatisplus.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateUtils;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author yaosang
 * @since 2021-09-29
 */
@Slf4j
@Service
public class KillCourseServiceImpl extends ServiceImpl<KillCourseMapper, KillCourse> implements IKillCourseService {

    @Autowired
    private RedissonClient redissonClient ;

    @Autowired
    private RedisTemplate<Object,Object> redisTemplate ;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private KillActivityMapper killActivityMapper;

    @Override
    public void add(KillCourse killCourse) {
        //1.判断参数
        AssertUtil.isNotNull(killCourse.getKillCount(),"秒杀数量不能为空");
        //省略...
        KillActivity killActivity = killActivityMapper.selectById(killCourse.getActivityId());
        AssertUtil.isNotNull(killActivity,"无效的秒杀课程");
        AssertUtil.isEquals(killActivity.getPublishStatus() , BaseConstants.KillActivity.STATUS_PUBLISH_WAIT,"活动场次已经发布,不可在加入新的秒杀课程");

        Wrapper<KillCourse> wrapper = new EntityWrapper<>();
        wrapper.eq("activity_id",killCourse.getActivityId());
        wrapper.eq("course_id",killCourse.getCourseId());
        KillCourse killCourseFromDB = super.selectOne(wrapper);
        AssertUtil.isNull(killCourseFromDB,"课程已经加入了该活动");

        //2.保存秒杀课程到数据库
        // 每个人只能秒杀1个
        killCourse.setPublishStatus(BaseConstants.Kill.STATUS_PUBLISH_WAIT);
        killCourse.setCreateTime(new Date());
        killCourse.setTeacherNames(killCourse.getTeacherNames());
        //时间戳，方便计算
        killCourse.setStartTime(killActivity.getBeginTime());
        killCourse.setEndTime(killActivity.getEndTime());
        //秒杀时间段
        killCourse.setTimeStr(killActivity.getTimeStr());

        insert(killCourse);
    }

    //定时任务每1小时执行一次
    public void publishKillActivity(){
        Date now = new Date() ;
        //提前12小时加入秒杀缓存
        Date endTime = DateUtils.addHours(now , 24);
        Wrapper<KillActivity> wrapper = new EntityWrapper<>();
        wrapper.eq("publish_status",BaseConstants.KillActivity.STATUS_PUBLISH_WAIT);
        //查询近12小时即将进入秒杀的
        //wrapper.lt("beginTime",endTime.getTime());
        List<KillActivity> killActivities = killActivityMapper.selectList(wrapper);
        if(killActivities.isEmpty())return;

        //转成Map，方便后面根据ID取对象
        Map<Long,KillActivity> killActivityMap = killActivities.stream().collect(Collectors.toMap(KillActivity::getId,KillActivity->KillActivity));

        //根据要发布的秒杀活动，查询出秒杀课程
        List<Long> killActivityIds = killActivities.stream().map(killActivity -> killActivity.getId()).collect(Collectors.toList());
        Wrapper<KillCourse> killCourseWrapper = new EntityWrapper<>();
        killCourseWrapper.in("activity_id",killActivityIds);
        List<KillCourse> killCourses = super.selectList(killCourseWrapper);
        if(killCourses.isEmpty())return;

        /**
         * 1.查询秒杀课程，条件：待发布，近2天的
         * 2.保存到Redis（hash结构）
         * 3.把库存提出来，使用信号量做库存预热
         * 4.把秒杀课程状态修改为，已经发布
         * 5.幂等处理，如果秒杀课程已经发布，不要在发布了
         */
        for(KillCourse killCourse: killCourses){
            //* 2.保存到Redis（hash结构）
            //* 5.幂等处理，如果秒杀课程已经发布，不要在发布了
            killCourse.setPublishStatus(BaseConstants.Kill.STATUS_PUBLISHING);
            killCourse.setPublishTime(now);

            //秒杀码
            String killCode = StrUtils.getComplexRandomString(16);
            killCourse.setKillCode(killCode);

            //使用putIfAbsent处理幂等
            boolean putSuccess = redisTemplate.opsForHash()
                    .putIfAbsent(BaseConstants.Kill.KEY_KILL_COURSES,killCourse.getId().toString(),killCourse);

            if(!putSuccess){
                log.warn("秒杀课程发布到Redis失败，可能已经被发布 -> {}" , killCourse);
                continue;
            }

            //* 3.把库存提出来，使用信号量做库存预热
            Integer killCount = killCourse.getKillCount();
            RSemaphore semaphore = redissonClient.getSemaphore(BaseConstants.Kill.KEY_KILL_STORE + killCourse.getId().toString());
            boolean setPermitsSuccess = semaphore.trySetPermits(killCount);
            if(!setPermitsSuccess){
                //库存发布到Redis失败，回滚
                redisTemplate.opsForHash().delete(BaseConstants.Kill.KEY_KILL_COURSES,killCourse.getId().toString());
                log.error("库存发布到Redis失败，课程从Redis中移除");
                continue;
            }

            //* 4.把秒杀课程状态修改为，已经发布

            updateById(killCourse);

            //活动
            KillActivity activity = killActivityMap.get(killCourse.getActivityId());
            activity.setPublishTime(now);
            activity.setPublishStatus(BaseConstants.KillActivity.STATUS_PUBLISHING);
            killActivityMapper.updateById(activity);
            log.info("秒杀课程发布到Redis {}",killCourse);
        }

    }



    @Override
    public KillCourse getOnlineKillCourseById(Long id) {
        return (KillCourse)redisTemplate.opsForHash().get(BaseConstants.Kill.KEY_KILL_COURSES, id.toString());
    }

    @Override
    public List getOnlineAllKillCourse() {
        return redisTemplate.boundHashOps(BaseConstants.Kill.KEY_KILL_COURSES).values();
    }

    /**
     * 1.判断
     *  1.1.判断ID不能为空
     *  1.2.等没登录
     *  1.3.是否已经秒杀过：同一个人，同一个课程，不能秒杀多次，除非支付成功
     *  1.4.秒杀时间合法
     *
     * 2.减去库存
     *  2.1.信号量减库存
     *
     * 3.下单
     *  3.1.创建订单号
     *  3.2.准备订单数据
     *  3.3.发送订单消息到MQ
     */
    public String kill(KillParamDto dto) {

        Long startTimeM = System.currentTimeMillis();

        //* 1.判断
        //*  1.1.判断ID不能为空
        AssertUtil.isNotNull(dto.getKillCourseId() , "无效的秒杀请求");
        AssertUtil.isNotNull(dto.getKillCode() , "无效的秒杀请求");

        //*  1.2.登没登录
        Long userId = LoginContext.getLogin().getId();

        Long killCourseId = dto.getKillCourseId();

        //*  1.3.是否已经秒杀过：同一个人，同一个课程，不能秒杀多次，除非支付成功
        Boolean isKilled = redisTemplate.hasKey(BaseConstants.Kill.KEY_KILLLOG + userId + ":" + killCourseId);

        //TODO 测试方便
        //AssertUtil.isFalse(isKilled,"您已经秒杀过了");

        //*  1.4.秒杀时间合法
        KillCourse killCourse = getOnlineKillCourseById(killCourseId);

        //秒杀码
        AssertUtil.isEquals(killCourse.getKillCode(),dto.getKillCode(),"无效的秒杀请求");

        Long nowTime = new Date().getTime();
        if(nowTime < killCourse.getStartTime() || nowTime > killCourse.getEndTime()){
            throw new GlobleException("无效的秒杀时间");
        }

        //* 2.减去库存
        //*  2.1.信号量减库存
        RSemaphore semaphore = redissonClient.getSemaphore(BaseConstants.Kill.KEY_KILL_STORE + killCourse.getId().toString());
        //减去库存
        boolean hasStore = semaphore.tryAcquire(dto.getKillCount());

        AssertUtil.isTrue(hasStore , "啊，你不是单身吧，手速不够快");

        //* 3.准备下单数据
        KilledInfo killedInfo = new KilledInfo();

        //*  3.1.创建订单号
        String orderNo = StrUtils.getRandomStringForTime()+StrUtils.getComplexRandomString(6);

        //*  3.2.准备订单数据
        killedInfo.setOrderNo(orderNo);
        killedInfo.setAmount(killCourse.getKillPrice());
        killedInfo.setUserId(userId);
        killedInfo.setKillCourseId(killCourseId);
        killedInfo.setKillCourseName(killCourse.getCourseName());
        killedInfo.setKillCoursePic(killCourse.getCoursePic());
        killedInfo.setKillCount(dto.getKillCount());//秒杀1个
        killedInfo.setCourseId(killCourse.getCourseId());

        //发送延迟消息，如果超时未下单支付，退库存
        Message<String> stringMessageBuilder = MessageBuilder.withPayload(JSON.toJSONString(killedInfo)).build();
        rocketMQTemplate.syncSend(
                BaseConstants.Kill.TOPIC_KILL_PAY_OUT_TIME+":"+BaseConstants.Kill.TAG_KILL_PAY_OUT_TIME,
                stringMessageBuilder,
                2000,6);   //模拟：10秒超时


        //*  3.3.保存秒杀信息到Redis
        redisTemplate.opsForValue().set(BaseConstants.Kill.KEY_KILLED_INFO+orderNo+":"+userId,killedInfo);

        //存放秒杀记录
        redisTemplate.opsForValue().set(BaseConstants.Kill.KEY_KILLLOG+userId+":"+killCourseId , "1");

        log.info("秒杀耗时：{}",System.currentTimeMillis() - startTimeM);
        return orderNo;
    }

    @Override
    public void removeKillCourseFromRedis() {
        BoundHashOperations<Object, String, KillCourse> boundHashOperations =
                redisTemplate.boundHashOps(BaseConstants.Kill.KEY_KILL_COURSES);

        List<KillCourse> values = boundHashOperations.values();

        Date now = new Date();

        if(values != null){

            values.forEach(killCourse -> {
                if(now.getTime()> killCourse.getEndTime()){

                    //秒杀课程下线
                    KillCourse killCourseFromDB = selectById(killCourse.getId());
                    killCourseFromDB.setPublishStatus(BaseConstants.Kill.STATUS_OFF_LINE);
                    updateById(killCourseFromDB);

                    //活动下线
                    KillActivity activity = killActivityMapper.selectById(killCourseFromDB.getActivityId());
                    activity.setPublishStatus(BaseConstants.KillActivity.STATUS_DOWN);
                    killActivityMapper.updateById(activity);

                    //移除缓存
                    boundHashOperations.delete(killCourse.getId().toString());
                    RSemaphore semaphore = redissonClient.getSemaphore(BaseConstants.Kill.KEY_KILL_STORE + killCourse.getId().toString());
                    semaphore.delete();

                    log.info("秒杀课程过期，从Redis移除...{}",killCourse);
                }
            });
        }
    }
}
